package com.lee.message.core.aspectj.rocketmq;

import com.lee.message.core.aspectj.AbstractMqIdempotentAspect;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;

/**
 * RocketMQ幂等性自动处理，放在事务注解内部执行，写入缓存失败回滚事务
 * 注：
 * 1.只处理集群模式重复消费，不处理广播模式重复消费
 * 2.不同消费组消费同一消息不受限制
 */
@Aspect
@Order(Ordered.LOWEST_PRECEDENCE)
@Slf4j
public class RocketMqIdempotentAspect extends AbstractMqIdempotentAspect {

    @Pointcut("@annotation(org.apache.rocketmq.spring.annotation.RocketMQMessageListener)"
            + "|| @within(org.apache.rocketmq.spring.annotation.RocketMQMessageListener)")
    public void idempotentPointCut() {
    }

    @Override
    public boolean isIdempotentCheck(ProceedingJoinPoint point) {
        // 广播模式不进行幂等性校验
        RocketMQMessageListener listener = getAnnotation(point, RocketMQMessageListener.class);
        MessageModel messageModel = listener.messageModel();
        if(messageModel.equals(MessageModel.BROADCASTING)){
            return false;
        }

        return true;
    }

    @Override
    public String consumerGroup(ProceedingJoinPoint point) {
        RocketMQMessageListener listener = getAnnotation(point, RocketMQMessageListener.class);
        return listener.consumerGroup();
    }

    @Override
    public String topic(ProceedingJoinPoint point) {
        RocketMQMessageListener listener = getAnnotation(point, RocketMQMessageListener.class);
        return listener.topic();
    }
}
